package rabbitmq.publisher.confirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * @ClassName PublisherConfirms
 * @Description 发布确认模式
 * @Author ZJX
 * @Date 2025/1/8 15:10
 * @Version 1.0
 **/
public class PublisherConfirms {

    public static final Integer MESSAGE_COUNT = 30000;

    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        return connectionFactory.newConnection();
    }

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //Strategy #1: Publishing Messages Individually   单独确认
        publishingMessagesIndividually();

        //Strategy #2: Publishing Messages in Batches   批量确认
        publishingMessagesInBatches();

        //Strategy #3: Handling Publisher Confirms Asynchronously 异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
//            1. 开启信道
            Channel channel = connection.createChannel();
//            2. 设置信道为confirm模式
            channel.confirmSelect();
//            3. 声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);
//            4. 监听confirm
            // 集合中存储的是未确认的消息ID
            long start = System.currentTimeMillis();
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSeqNo.remove(deliveryTag);
                    }
//                    根据业务逻辑操作 比如重发
                }
            });
//            5. 发送消息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher confirms" + i;
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());
                confirmSeqNo.add(seqNo);
            }
            while (!confirmSeqNo.isEmpty()){
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.printf("异步确认策略, 消息条数 : %d, 耗时: %d ms%n", MESSAGE_COUNT, end - start);
        }
    }

    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
//            1. 开启信道
            Channel channel = connection.createChannel();
//            2. 设置信道为confirm模式
            channel.confirmSelect();
//            3. 声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);
//            4.发送消息并进行确认
            int batchSize = 100;
            int outStandingMessageCount = 0;
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "单独确认策略" + i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2,null,message.getBytes(StandardCharsets.UTF_8));
                outStandingMessageCount++;
                if (outStandingMessageCount == batchSize){
                    channel.waitForConfirms(5000);
                    outStandingMessageCount = 0;
                }
            }
            if (outStandingMessageCount > 0){
                channel.waitForConfirms(5000);
            }

            long end = System.currentTimeMillis();
            System.out.printf("批量确认策略, 消息条数 : %d, 耗时: %d ms%n", MESSAGE_COUNT, end - start);
        }
    }



        private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
//            1. 开启信道
            Channel channel = connection.createChannel();
//            2. 设置信道为confirm模式
            channel.confirmSelect();
//            3. 声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
//            4. 发送消息并等待确认
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "单独确认策略" + i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,message.getBytes(StandardCharsets.UTF_8));
//                等待确认
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("单独确认策略, 消息条数 : %d, 耗时: %d ms%n", MESSAGE_COUNT, end - start);
        }
    }
}
